package com.song.stormdemo1;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class DataSourceSpout extends BaseRichSpout {
	
	
	private Map conf;
	private TopologyContext context;
	private SpoutOutputCollector collector;
	
	public void nextTuple() {
		Collection<File> files = FileUtils.listFiles(new File("D:\\software\\test"), new String[]{"txt"}, true);
		if(files != null && files.size() >0)
		{
			try {
				for(File file :files)
				{
					List<String> lines = FileUtils.readLines(file);	
					for(String line : lines)
					{
						collector.emit(new Values(line));
					}
			        FileUtils.moveFile(file, new File(file.getAbsolutePath() + "." + System.currentTimeMillis())); 
				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		this.conf = conf;
		this.context =context;
		this.collector = collector;

	}																								

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("line"));
		
	}

}
